package com.hrpc.proxy.handler;

import com.hrpc.IdGenerate;
import com.hrpc.NettyBootStrapInitializer;
import com.hrpc.annotation.TryTimes;
import com.hrpc.compress.CompressorFactory;
import com.hrpc.discovery.Register;
import com.hrpc.enumeration.RequestType;
import com.hrpc.exceptions.NetworkException;
import com.hrpc.hRpcBootStarp;
import com.hrpc.protection.CircuitBreaker;
import com.hrpc.serialize.SerializeFactory;
import com.hrpc.transport.message.RequestPayload;
import com.hrpc.transport.message.hRpcRequest;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 *  该类封装了客户端通信的基础逻辑，每一个代理对象的远程调用过程都封装在了invoke方法中
 *  1.发现可用服务
 *  2.建立连接
 *  3.发送请求
 *  4.得到结果
 */
public class RpcClientInvocationHandler implements InvocationHandler {
    //此处需要一个注册中心和一个接口
    private Register register;
    private Class<?> interfaceRef;
    private String group;


    public RpcClientInvocationHandler(Register register, Class<?> interfaceRef,String group) {
        this.register = register;
        this.interfaceRef = interfaceRef;
        this.group=group;
    }

    public RpcClientInvocationHandler() {
    }

    /**
     * 所有的方法调用，本质都会走这里
     * @param proxy the proxy instance that the method was invoked on
     *
     * @param method the {@code Method} instance corresponding to
     * the interface method invoked on the proxy instance.  The declaring
     * class of the {@code Method} object will be the interface that
     * the method was declared in, which may be a superinterface of the
     * proxy interface that the proxy class inherits the method through.
     *
     * @param args an array of objects containing the values of the
     * arguments passed in the method invocation on the proxy instance,
     * or {@code null} if interface method takes no arguments.
     * Arguments of primitive types are wrapped in instances of the
     * appropriate primitive wrapper class, such as
     * {@code java.lang.Integer} or {@code java.lang.Boolean}.
     *
     * @return
     * @throws Throwable
     */
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        CircuitBreaker circuitBreaker=new CircuitBreaker(5,0.5F);

        //从接口中判断是否需要重试
        TryTimes tryTimeAnnotation = method.getAnnotation(TryTimes.class);
        //默认值代表不重试，拿到注解代表要充实
        int tryTimes=0;
        int intervalTime=2000;
        if (tryTimeAnnotation != null){
            tryTimes=tryTimeAnnotation.tryTimes();
            intervalTime=tryTimeAnnotation.intervalTime();
        }
        while (true) {
            //什么情况下需要重试？1、异常 2、响应有问题 code==500
            try {

                //我们调用sayHi方法，事实上会走进这个代码段中
                //我们已经知道method，同时也知道args
                /**
                 * 1.封装报文
                 */
                RequestPayload requestPayload=new RequestPayload(interfaceRef.getName(),
                        method.getName(), method.getParameterTypes(),
                        args,method.getReturnType());
                // 需要对各种请求ID和各种请求类型做处理
                hRpcRequest hRpcRequest=new hRpcRequest(
                        hRpcBootStarp.getInstance().getConfiguration().getId_generate().getID(),
                        RequestType.REQUEST.getId(),
                        CompressorFactory.getCompressor(hRpcBootStarp.getInstance().getConfiguration().getCompressType()).getCode(),
                        SerializeFactory.getSerialize(hRpcBootStarp.getInstance().getConfiguration().getSerializeType()).getCode(),
                        requestPayload);
                /**
                 * 2
                 */
                // 将请求存入本地线程，需要在合适的时候调用remove方法
                hRpcBootStarp.REQUEST_THREAD_LOCAL.set(hRpcRequest);

                /**
                 * 3、做服务发现
                 */
                //2.从注册中心拉取服务列表，并通过客户端负载均衡寻找一个可用的服务
                //传入服务的名字，返回一个IP+端口
                //1.尝试获取当前配置的负载均衡器，选取一个可用节点
                InetSocketAddress address = hRpcBootStarp.getInstance().getConfiguration().getLoad_balancer().selectServerAddress(interfaceRef.getName(),group);
                //.使用netty连接服务器，发送调用的服务的名字+方法名+参数列表，得到结果
                //定义线程池，EventLoopGroup
                //Q：整个连接过程放在这里行不行？每次调用都会产生一个新的netty连接，如何缓存我们的连接
                //也就意味着在此处建立一个新的连接是不合适的
                //获取当前地址所对应的断路器
                boolean aBreak = circuitBreaker.isBreak();
                Map<SocketAddress, CircuitBreaker> everyIpCircuitBreaker =
                        hRpcBootStarp.getInstance().getConfiguration().getEveryIpCircuitBreaker();

                /**
                 * 4.获取当前地址对应的熔断器
                 */
                CircuitBreaker circuitBreaker1 = everyIpCircuitBreaker.get(address);
                if (circuitBreaker1 == null){
                    circuitBreaker1=new CircuitBreaker(10,0.5F);
                    everyIpCircuitBreaker.put(address,circuitBreaker1);
                }
                //如果断路器是打开的
                if (hRpcRequest.getRequestType()!=RequestType.HEART_BEAT.getId() &&
                        aBreak == true){
                    //断路器打来说明当前请求不应该再发送
                    //定期打开
                    Timer timer=new Timer();
                    timer.schedule(new TimerTask() {
                        @Override
                        public void run() {
                            hRpcBootStarp.getInstance().getConfiguration().getEveryIpCircuitBreaker()
                                    .get(address).reSet();
                        }
                    },5000);
                    throw new RuntimeException("当前断路器已经开启，无法发送请求！！！");
                }
                //解决方案：缓存channel，尝试从缓存中获取channel，如果未获取则创建新的连接并进行缓存，缓存到hRpcBootStarp中
                /**
                 * 5.尝试获取一个可用通道
                 */
                Channel channel=getAvailableChannel(address);


                /**
                 * --------------同步策略（不合适）--------------------
                 */
    //                ChannelFuture channelFuture = channel.writeAndFlush(new Object()).await();
    //                //需要学习channelFuture一些简单的API，getNow获取当前的返回结果，如果未处理完成返回null
    //                if (channelFuture.isDone()){
    //                    Object object = channelFuture.getNow();
    //                }else if(!channelFuture.isSuccess()){
    //                    //需要捕获异常，可以捕获异步任务中的异常
    //                    Throwable cause = channelFuture.cause();
    //                    throw new RuntimeException(cause);
    //                }
                /**
                 * -----------------异步策略---------------------
                 */
                /**
                 * 6、写出报文
                 */
                CompletableFuture<Object> completableFuture=new CompletableFuture<>();
                // 需要将completableFuture暴露出去
                hRpcBootStarp.PENDING_REQUEST.put(hRpcRequest.getRequestId(), completableFuture);
                //这里直接writeandFlush写出了一个请求，这个请求的实例就会进入pipeline执行出站的一系列操作
                //实际上第一个出站程序一定是将hRpcRequest转换为二进制报文

                channel.writeAndFlush(hRpcRequest).addListener((ChannelFutureListener) promise->{
        //                    if (promise.isDone()){
        //                        //当前promise将来的返回的结果是writeAndFlush的结果
        //                        //一旦数据被写进去，这个promise也就结束了
        //                        //但是我们想要的是服务端给我们的返回结果，所以这里处理completableFuture是有问题的
        //                        //应该将completableFuture挂起并暴露，并且在得到服务提供方的响应时调用complete方法
        //                        completableFuture.complete(promise.getNow());
        //                    }
                    //只需要处理异常就好了
                    if (!promise.isSuccess()) {
                        completableFuture.completeExceptionally(promise.cause());
                    }
                });
                /**
                 * 7、清理threadlocal
                 */
                hRpcBootStarp.REQUEST_THREAD_LOCAL.remove();

    //              如果没有地方处理completableFuture，这里会阻塞，等到complete方法的执行
                //那我们需要在那里调用complete方法得到结果，很明显是在pipLine中最终的handler的处理结果
                /**
                 * 8.获得响应的结果，并返回
                 */
                Object result = completableFuture.get(100, TimeUnit.SECONDS);
                //记录成功的请求，
                circuitBreaker.recordRequest();
                return result;
            } catch (Exception e) {
                //次数减一并且等待固定时间，固定时间有一定的问题--->重试风暴
                tryTimes--;
                //记录错误的次数
                circuitBreaker.recordErrorRequest();
                Thread.sleep(intervalTime);
                if (tryTimes <= 0){
                    break;
                }
            }
        }
        throw new RuntimeException("执行远程方法调用时出现异常！！！");
    }

    /**
     * 根据地址获取一个可用的通道
     * @param address
     * @return
     */
    private Channel getAvailableChannel(InetSocketAddress address) {
        //1.尝试从缓存中获取channel
        Channel channel = hRpcBootStarp.CHANNEL_CACHE.get(address);
        //拿不到就去建立连接
        if (channel == null){
            //await()会阻塞等待连接成功再返回，netty提供了异步处理的逻辑
            //sync()和await()都是阻塞当前线程获取返回值（因为连接的过程是异步的，发送数据的过程是异步的）
            //如果发生了异常，sync会主动在主线程抛出异常，await()不会，await()在子线程中处理需要使用future
            //同步
//                    channel = NettyBootStrapInitializer.getBootstrap().connect(address).await().channel();
            //异步：使用addListener()执行的异步操作
            CompletableFuture<Channel> channelFuture=new CompletableFuture<>();
            NettyBootStrapInitializer.getBootstrap().connect(address).addListener(
                    (ChannelFutureListener) promise->{
                        if (promise.isDone()){
                            //异步的，我们已经完成
                            channelFuture.complete(promise.channel());
                        } else if (!promise.isSuccess()) {
                            channelFuture.completeExceptionally(promise.cause());
                        }
                    }
            );
            //阻塞获取chan
            try {
                channel = channelFuture.get(3, TimeUnit.SECONDS);
            } catch (InterruptedException | TimeoutException | ExecutionException e) {
                throw new RuntimeException("获取通道时发生异常！！！");
            }
            //缓存channel本身
            hRpcBootStarp.CHANNEL_CACHE.put(address,channel);
        }
        if (channel == null){
            throw new NetworkException("获取channel时发生了异常！！！");
        }
        return channel;

    }
}
